{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "chapter-intro",
   "metadata": {},
   "source": [
    "# 第 6 章：构建 AI 智能体的其它 API 选项\n",
    "\n",
    "> 本笔记文件需要与《LangGraph实战》的第 6 章的内容配套使用。\n",
    "\n",
    "在构建智能体的道路上，框架赋予我们多种途径，如同百川汇流，殊途同归。关键在于洞悉每条路径的特性，权衡利弊，最终择取那条最能释放创造力、最能达成目标的通途。\n",
    "\n",
    "本章将深入探索 LangGraph 框架的核心 —— 其丰富而强大的 API 工具箱。为了满足不同开发场景和开发者偏好，LangGraph 提供了多种 API 接口：\n",
    "\n",
    "- **`create_react_agent`**: 预构建 API，快速启动按钮，迅速搭建功能完备的 ReAct 智能体\n",
    "- **Functional API**: 以函数为中心的 API，通过 `@entrypoint` 和 `@task` 装饰器，将 LangGraph 的核心功能融入熟悉的函数式编程范式\n",
    "- **Graph API**: 核心 API 范式，通过显式定义节点和边构建任意复杂度的智能体架构"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "environment-setup",
   "metadata": {},
   "source": [
    "### 🚀 环境准备\n",
    "\n",
    "首先加载必要的环境变量配置："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cell-0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from dotenv import load_dotenv\n",
    "\n",
    "load_dotenv()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "section-6-1",
   "metadata": {},
   "source": [
    "## 6.1 create_react_agent \n",
    "\n",
    "LangGraph 作为 AI 智能体开发框架，在灵活性和功能性方面展现出独特优势。该框架不仅支持高度定制化的智能体架构开发，同时为快速原型设计和开发者入门提供了便捷工具。其中，`create_react_agent` 函数作为预构建组件，实现了 ReAct（推理—\n",
    "行动）模式的快速部署，显著降低了开发门槛。本节将演示 `create_react_agent` 的核心功能、自定义选项，以及如何在更广泛的 LangChain 生态系统中无缝集成。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-1",
   "metadata": {},
   "source": [
    "##### 示例 6-1：`create_react_agent` 的基本用法\n",
    "\n",
    "让我们从最简单的示例开始，构建一个能够查询天气信息的智能体："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cell-1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "The weather in San Francisco is currently sunny! 🌞 While the city is known for its foggy days, it's great to hear it's clear up today. Let me know if you'd like more details!\n"
     ]
    }
   ],
   "source": [
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.prebuilt import create_react_agent\n",
    "from langchain_core.tools import tool\n",
    "\n",
    "# 初始化语言模型\n",
    "model = ChatOpenAI(model=\"Qwen/Qwen3-8B\", temperature=0)\n",
    "\n",
    "# 定义一个简单的工具来获取天气信息\n",
    "@tool\n",
    "def get_weather(city: str):\n",
    "    \"\"\"Use this tool to get the weather information for a city.\"\"\"\n",
    "    if city.lower() in [\"nyc\", \"new york\"]:\n",
    "        return \"It might be cloudy in New York.\"\n",
    "    elif city.lower() in [\"sf\", \"san francisco\"]:\n",
    "        return \"It's always sunny in San Francisco.\"\n",
    "    else:\n",
    "        return \"Unable to get the weather information for this city.\"\n",
    "\n",
    "tools = [get_weather]\n",
    "\n",
    "# 创建 ReAct 智能体\n",
    "graph = create_react_agent(model, tools=tools)\n",
    "\n",
    "# 调用智能体\n",
    "inputs = {\"messages\": [(\"user\", \"How's the weather in sf?\")]}\n",
    "response = graph.invoke(inputs)\n",
    "print(response['messages'][-1].content)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-1",
   "metadata": {},
   "source": [
    "**💡 核心概念解析**：\n",
    "\n",
    "- **模型绑定**: `create_react_agent` 自动将工具绑定到语言模型\n",
    "- **工具定义**: 使用 `@tool` 装饰器将 Python 函数转换为 LangChain 工具\n",
    "- **图构建**: 内部自动构建 ReAct 工作流程图，包含 agent 和 tools 节点\n",
    "- **消息处理**: 支持标准的消息格式输入和输出"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-2",
   "metadata": {},
   "source": [
    "##### 示例 6-2：使用自定义系统提示\n",
    "\n",
    "通过 `prompt` 参数，我们可以自定义智能体的行为和响应风格："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "cell-2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "紐約目前可能有雲，天氣較為陰沉。\n"
     ]
    }
   ],
   "source": [
    "# 定义用于中文回复的自定义系统提示\n",
    "chinese_prompt = \"请用中文回复所有问题\"\n",
    "\n",
    "# 使用自定义提示创建 ReAct 智能体\n",
    "graph_chinese = create_react_agent(model, tools=tools, prompt=chinese_prompt)\n",
    "\n",
    "# 示例调用\n",
    "inputs = {\"messages\": [(\"user\", \"How's the weather in nyc?\")]}\n",
    "response_chinese = graph_chinese.invoke(inputs)\n",
    "print(response_chinese['messages'][-1].content)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-2",
   "metadata": {},
   "source": [
    "**💡 提示定制的价值**：\n",
    "\n",
    "- **语言控制**: 指定回复语言\n",
    "- **角色设定**: 定义智能体的角色和个性\n",
    "- **行为约束**: 设置特定的行为规则和限制\n",
    "- **领域适应**: 为特定应用领域提供上下文"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-3",
   "metadata": {},
   "source": [
    "##### 示例 6-3：添加对话记忆功能\n",
    "\n",
    "通过 `checkpointer` 参数，智能体可以跨多轮对话维护上下文："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "cell-3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "The current weather in San Francisco is sunny. 🌞 San Francisco is known for its mild climate and occasional fog, but today the skies are clear. Hope you enjoy the sunshine!\n",
      "\n",
      "\n",
      "I'm sorry, but I couldn't retrieve the weather information for Chicago. It's possible there was a temporary issue or the city name isn't recognized. Let me know if you'd like to check another city or need further assistance! ☁️☀️\n"
     ]
    }
   ],
   "source": [
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.prebuilt import create_react_agent\n",
    "from langgraph.checkpoint.memory import MemorySaver\n",
    "\n",
    "# ... 定义模型和工具（与之前相同）...\n",
    "\n",
    "# 初始化内存检查点\n",
    "memory = MemorySaver()\n",
    "\n",
    "# 创建具有记忆功能的 ReAct 智能体\n",
    "graph_with_memory = create_react_agent(model, tools=tools, checkpointer=memory)\n",
    "\n",
    "# 首次交互\n",
    "config = {\"configurable\": {\"thread_id\": \"user_thread_1\"}} # 唯一线程 ID\n",
    "inputs_1 = {\"messages\": [(\"user\", \"How's the weather in sf?\")]}\n",
    "response_1 = graph_with_memory.invoke(inputs_1, config=config)\n",
    "print(response_1['messages'][-1].content)\n",
    "\n",
    "# 同一线程中的第二次交互 - 智能体记住上下文\n",
    "inputs_2 = {\"messages\": [(\"user\", \"How is chicago?\")]}\n",
    "response_2 = graph_with_memory.invoke(inputs_2, config=config)\n",
    "print(response_2['messages'][-1].content)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-3",
   "metadata": {},
   "source": [
    "**💡 记忆机制解析**：\n",
    "\n",
    "- **线程隔离**: 通过 `thread_id` 区分不同的对话会话\n",
    "- **状态持久化**: `MemorySaver` 在内存中保存对话历史\n",
    "- **上下文连续性**: 智能体能够理解跨轮次的对话上下文\n",
    "- **存储选择**: 生产环境可选择 `RedisSaver` 或 `SQLiteSaver`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-4",
   "metadata": {},
   "source": [
    "##### 示例 6-4：实现人机环路工作流程\n",
    "\n",
    "通过 `interrupt_before` 参数，我们可以在关键节点暂停智能体执行，实现人工监督："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "cell-4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "执行智能体（将在工具调用前暂停）...\n",
      "消息类型: HumanMessage\n",
      "内容: How's the weather in sf?\n",
      "消息类型: AIMessage\n",
      "计划调用工具: [{'name': 'get_weather', 'args': {'city': 'San Francisco'}, 'id': '0198ebdf1c7d7d658f8d063b41a21d79', 'type': 'tool_call'}]\n",
      ">>> 智能体已暂停，等待人工审核 <<<\n",
      "\n",
      "当前智能体状态:\n",
      "下一个节点: ('tools',)\n",
      "消息数量: 2\n"
     ]
    }
   ],
   "source": [
    "# 创建启用人机环路的 ReAct 智能体 - 在工具调用前中断\n",
    "graph_hitl = create_react_agent(model, tools=tools, interrupt_before=[\"tools\"], checkpointer=memory)\n",
    "\n",
    "# 首次交互 - 智能体将在工具调用之前暂停\n",
    "config_hitl = {\"configurable\": {\"thread_id\": \"user_thread_hitl\"}}\n",
    "inputs_hitl = {\"messages\": [(\"user\", \"How's the weather in sf?\")]}\n",
    "\n",
    "print(\"执行智能体（将在工具调用前暂停）...\")\n",
    "stream = graph_hitl.stream(inputs_hitl, config=config_hitl, stream_mode=\"values\")\n",
    "for output in stream:\n",
    "    if 'messages' in output and output['messages']:\n",
    "        last_message = output['messages'][-1]\n",
    "        print(f\"消息类型: {type(last_message).__name__}\")\n",
    "        if hasattr(last_message, 'tool_calls') and last_message.tool_calls:\n",
    "            print(f\"计划调用工具: {last_message.tool_calls}\")\n",
    "            print(\">>> 智能体已暂停，等待人工审核 <<<\")\n",
    "            break\n",
    "        else:\n",
    "            print(f\"内容: {last_message.content}\")\n",
    "\n",
    "# 检查当前状态\n",
    "print(\"\\n当前智能体状态:\")\n",
    "current_state = graph_hitl.get_state(config_hitl)\n",
    "print(f\"下一个节点: {current_state.next}\")\n",
    "print(f\"消息数量: {len(current_state.values['messages'])}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-4",
   "metadata": {},
   "source": [
    "**💡 人机环路的核心价值**：\n",
    "\n",
    "- **安全控制**: 在执行敏感操作前引入人工审核\n",
    "- **质量保证**: 确保工具调用的准确性和合理性\n",
    "- **风险管理**: 防止自动化系统执行不当操作\n",
    "- **学习机会**: 通过人工反馈改进智能体决策"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-5",
   "metadata": {},
   "source": [
    "##### 示例 6-5：结构化输出响应\n",
    "\n",
    "使用 `response_format` 参数，智能体可以返回符合预定义结构的响应："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cell-5",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydantic import BaseModel, Field\n",
    "from langchain_openai import ChatOpenAI\n",
    "\n",
    "# 定义结构化输出的 Pydantic 模型\n",
    "class WeatherResponse(BaseModel):\n",
    "    \"\"\"Respond with a weather description.\"\"\"\n",
    "    city: str = Field(description=\"The city name\")\n",
    "    conditions: str = Field(description=\"Weather conditions description\")\n",
    "    temperature: str = Field(description=\"Temperature information\")\n",
    "    confidence: str = Field(description=\"Confidence level of the weather data\")\n",
    "\n",
    "# 注意：结构化输出功能需要支持原生结构化输出的模型\n",
    "# 这里我们演示概念，实际使用时请确保模型支持，建议使用 OpenAI 模型实验\n",
    "try:\n",
    "    # 创建启用结构化输出的 ReAct 智能体\n",
    "    graph_structured = create_react_agent(model, tools=tools, response_format=WeatherResponse)\n",
    "    \n",
    "    # 示例调用\n",
    "    inputs_structured = {\"messages\": [(\"user\", \"How's the weather in sf?\")]}\n",
    "    response_structured = graph_structured.invoke(inputs_structured)\n",
    "    \n",
    "    # 访问结构化响应\n",
    "    if \"structured_response\" in response_structured:\n",
    "        structured_data = response_structured[\"structured_response\"]\n",
    "        print(\"结构化天气响应:\")\n",
    "        print(f\"城市: {structured_data.city}\")\n",
    "        print(f\"天气状况: {structured_data.conditions}\")\n",
    "        print(f\"温度: {structured_data.temperature}\")\n",
    "        print(f\"置信度: {structured_data.confidence}\")\n",
    "    else:\n",
    "        print(\"常规响应:\", response_structured['messages'][-1].content)\n",
    "except Exception as e:\n",
    "    print(f\"结构化输出示例跳过 (需要兼容模型): {e}\")\n",
    "    print(\"使用常规输出进行演示...\")\n",
    "    regular_response = graph.invoke({\"messages\": [(\"user\", \"How's the weather in sf?\")]})\n",
    "    print(regular_response['messages'][-1].content)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-5",
   "metadata": {},
   "source": [
    "**💡 结构化输出的应用场景**：\n",
    "\n",
    "- **API 集成**: 与下游系统的标准化数据交换\n",
    "- **数据处理**: 便于程序化处理和分析\n",
    "- **用户界面**: 为前端提供结构化的展示数据\n",
    "- **质量控制**: 确保输出格式的一致性和完整性"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "section-6-2",
   "metadata": {},
   "source": [
    "## 6.2 Functional API：拥抱工作流设计的简洁性和熟悉性\n",
    "\n",
    "Functional API 作为 Graph API 的补充，旨在提供一种更简洁、更直观、更贴近传统编程习惯的工作流构建方式。它在保留 LangGraph 核心功能的同时，让开发者能够使用熟悉的 Python 函数和装饰器来定义工作流。\n",
    "\n",
    "### 设计理念\n",
    "- **降低学习门槛**: 对新手开发人员更加友好\n",
    "- **代码集成性**: 与现有 Python 代码库无缝集成\n",
    "- **快速开发**: 减少样板代码，加速原型设计\n",
    "- **功能完整性**: 支持持久性、内存、人机环路和流式传输\n",
    "\n",
    "### 核心组件：`@entrypoint` 和 `@task`\n",
    "\n",
    "Functional API 基于两个基本装饰器构建：\n",
    "- **`@entrypoint`**: 定义工作流的入口点和边界\n",
    "- **`@task`**: 封装独立的工作单元"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-6",
   "metadata": {},
   "source": [
    "##### 示例 6-6：具有参数用法的复杂 `@entrypoint` 函数\n",
    "\n",
    "展示 `@entrypoint` 装饰器的高级用法，包括所有可注入参数："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "cell-6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 使用 invoke 执行 ===\n",
      "结果: 工作流处理的输入：{'message': 'Hello'}, 执行次数: 1\n",
      "\n",
      "=== 使用 stream 执行 ===\n",
      "流式数据: ('custom', \"工作流 'complex_workflow_1' 已启动\")\n",
      "流式数据: ('custom', \"发现先前状态: 工作流处理的输入：{'message': 'Hello'}, 执行次数: 1\")\n",
      "流式数据: ('custom', '更新计数器: 2')\n",
      "流式数据: ('custom', '工作流处理完成')\n",
      "流式数据: ('updates', {'my_workflow': \"工作流处理的输入：{'message': 'World'}, 执行次数: 2\"})\n"
     ]
    }
   ],
   "source": [
    "from langgraph.func import entrypoint\n",
    "from langgraph.checkpoint.memory import MemorySaver\n",
    "from langgraph.store.memory import InMemoryStore, BaseStore\n",
    "from langgraph.types import StreamWriter\n",
    "from langchain_core.runnables import RunnableConfig\n",
    "from typing import Any\n",
    "\n",
    "store = InMemoryStore()\n",
    "checkpointer = MemorySaver()\n",
    "\n",
    "@entrypoint(checkpointer=checkpointer, store=store)\n",
    "def my_workflow(\n",
    "    user_input: dict, # 输入参数\n",
    "    *,\n",
    "    previous: Any = None, # 用于先前状态的可注入参数\n",
    "    store: BaseStore,    # 用于长期内存存储的可注入参数\n",
    "    writer: StreamWriter, # 用于自定义流写入器的可注入参数\n",
    "    config: RunnableConfig # 用于运行时配置的可注入参数\n",
    ") -> str:\n",
    "    \"\"\"一个复杂的演示入口点参数的工作流。\"\"\"\n",
    "    thread_id = config.get(\"configurable\", {}).get(\"thread_id\", \"unknown\")\n",
    "    \n",
    "    # 向客户端流式传输自定义信息\n",
    "    writer(f\"工作流 '{thread_id}' 已启动\")\n",
    "    \n",
    "    # 检查先前状态\n",
    "    if previous:\n",
    "        writer(f\"发现先前状态: {previous}\")\n",
    "    else:\n",
    "        writer(\"这是首次执行\")\n",
    "    \n",
    "    # 访问长期存储\n",
    "    try:\n",
    "        stored_data = store.get((\"demo\", \"counter\"), \"value\")\n",
    "        if stored_data:\n",
    "            counter = stored_data.value + 1\n",
    "        else:\n",
    "            counter = 1\n",
    "        store.put((\"demo\", \"counter\"), \"value\", counter)\n",
    "        writer(f\"更新计数器: {counter}\")\n",
    "    except Exception as e:\n",
    "        writer(f\"存储操作错误: {e}\")\n",
    "        counter = 1\n",
    "    \n",
    "    result = f\"工作流处理的输入：{user_input}, 执行次数: {counter}\"\n",
    "    writer(\"工作流处理完成\")\n",
    "    \n",
    "    return result\n",
    "\n",
    "# 示例调用\n",
    "config = {\"configurable\": {\"thread_id\": \"complex_workflow_1\"}}\n",
    "print(\"=== 使用 invoke 执行 ===\")\n",
    "result = my_workflow.invoke({\"message\": \"Hello\"}, config)\n",
    "print(f\"结果: {result}\")\n",
    "print()\n",
    "\n",
    "# 使用流式执行查看中间过程\n",
    "print(\"=== 使用 stream 执行 ===\")\n",
    "for chunk in my_workflow.stream({\"message\": \"World\"}, config, stream_mode=[\"custom\", \"updates\"]):\n",
    "    print(f\"流式数据: {chunk}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-6",
   "metadata": {},
   "source": [
    "**💡 `@entrypoint` 高级特性解析**：\n",
    "\n",
    "- **可注入参数**: `previous`、`store`、`writer`、`config` 由运行时自动注入\n",
    "- **状态持久化**: 通过 `checkpointer` 参数实现跨执行状态保存\n",
    "- **长期存储**: `store` 参数提供持久化存储能力\n",
    "- **实时流式传输**: `writer` 支持自定义数据流式输出\n",
    "- **运行时配置**: `config` 提供执行时的配置信息访问"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-7",
   "metadata": {},
   "source": [
    "##### 示例 6-7：带有重试策略的 `@task` 函数\n",
    "\n",
    "演示如何使用 `@task` 装饰器的重试功能来处理不可靠的操作："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "cell-7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "测试带重试策略的任务...\n",
      "尝试从 https://api.example.com/data 获取数据...\n",
      "API 请求超时！\n",
      "尝试从 https://api.example.com/data 获取数据...\n",
      "API 请求超时！\n",
      "尝试从 https://api.example.com/data 获取数据...\n",
      "API 请求成功！\n",
      "\n",
      "最终结果: {'workflow_result': '数据已处理', 'api_response': {'status': 'success', 'data': '来自 https://api.example.com/data 的数据'}}\n"
     ]
    }
   ],
   "source": [
    "from langgraph.func import task\n",
    "from langgraph.types import RetryPolicy\n",
    "import time\n",
    "import random\n",
    "\n",
    "# 定义重试策略\n",
    "retry_policy = RetryPolicy(max_attempts=3, retry_on=TimeoutError)\n",
    "\n",
    "@task(name=\"api_data_fetcher\", retry_policy=retry_policy)\n",
    "def fetch_api_data(api_endpoint: str) -> dict:\n",
    "    \"\"\"从带有重试策略的 API 端点获取数据的任务。\"\"\"\n",
    "    print(f\"尝试从 {api_endpoint} 获取数据...\")\n",
    "    time.sleep(random.uniform(0.1, 0.5)) # 模拟网络延迟\n",
    "    \n",
    "    if random.random() < 0.8: # 模拟 60% 的超时概率\n",
    "        print(\"API 请求超时！\")\n",
    "        raise TimeoutError(\"API 请求超时\")\n",
    "    \n",
    "    print(\"API 请求成功！\")\n",
    "    return {\"status\": \"success\", \"data\": f\"来自 {api_endpoint} 的数据\"}\n",
    "\n",
    "# 在入口点内调用任务的示例\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def data_processing_workflow(endpoint_url: str) -> dict:\n",
    "    \"\"\"调用 fetch_api_data 任务的工作流。\"\"\"\n",
    "    try:\n",
    "        api_result = fetch_api_data(api_endpoint=endpoint_url).result()\n",
    "        return {\"workflow_result\": \"数据已处理\", \"api_response\": api_result}\n",
    "    except TimeoutError as e:\n",
    "        return {\"workflow_result\": \"处理失败\", \"error\": str(e)}\n",
    "\n",
    "# 测试重试机制\n",
    "config = {\"configurable\": {\"thread_id\": \"task_params_workflow_1\"}}\n",
    "print(\"测试带重试策略的任务...\")\n",
    "result = data_processing_workflow.invoke(\"https://api.example.com/data\", config)\n",
    "print(f\"\\n最终结果: {result}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-7",
   "metadata": {},
   "source": [
    "**💡 `@task` 重试机制的优势**：\n",
    "\n",
    "- **容错能力**: 自动处理瞬时网络错误和服务不可用\n",
    "- **指定异常**: `retry_on` 参数允许针对特定异常类型重试\n",
    "- **自定义策略**: 可配置最大重试次数和重试条件\n",
    "- **任务隔离**: 重试逻辑封装在任务级别，不影响整体工作流"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-8",
   "metadata": {},
   "source": [
    "##### 示例 6-8：包含控制流和任务调用的工作流逻辑\n",
    "\n",
    "展示在 `@entrypoint` 中使用 Python 控制流和任务调用构建条件逻辑："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "cell-8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 测试偶数 ===\n",
      "处理数字: 4\n",
      "检查 4 是否为偶数...\n",
      "将 4 乘以 2...\n",
      "结果: {'input': 4, 'is_even': True, 'operation': '乘以2', 'result': 8}\n",
      "\n",
      "=== 测试奇数 ===\n",
      "处理数字: 5\n",
      "检查 5 是否为偶数...\n",
      "将 5 加 10...\n",
      "结果: {'input': 5, 'is_even': False, 'operation': '加10', 'result': 15}\n"
     ]
    }
   ],
   "source": [
    "@task\n",
    "def is_even(number: int) -> bool:\n",
    "    \"\"\"检查数字是否为偶数的任务。\"\"\"\n",
    "    print(f\"检查 {number} 是否为偶数...\")\n",
    "    return number % 2 == 0\n",
    "\n",
    "@task\n",
    "def multiply_by_two(number: int) -> int:\n",
    "    \"\"\"将数字乘以二的任务。\"\"\"\n",
    "    print(f\"将 {number} 乘以 2...\")\n",
    "    return number * 2\n",
    "\n",
    "@task\n",
    "def add_ten(number: int) -> int:\n",
    "    \"\"\"将数字加十的任务。\"\"\"\n",
    "    print(f\"将 {number} 加 10...\")\n",
    "    return number + 10\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def number_workflow(input_number: int) -> dict:\n",
    "    \"\"\"根据数字是奇数还是偶数处理数字的工作流。\"\"\"\n",
    "    print(f\"处理数字: {input_number}\")\n",
    "    \n",
    "    # 检查是否为偶数\n",
    "    is_even_result = is_even(input_number).result()\n",
    "    \n",
    "    if is_even_result:\n",
    "        # 偶数：乘以2\n",
    "        result = multiply_by_two(input_number).result()\n",
    "        operation = \"乘以2\"\n",
    "    else:\n",
    "        # 奇数：加10\n",
    "        result = add_ten(input_number).result()\n",
    "        operation = \"加10\"\n",
    "    \n",
    "    return {\n",
    "        \"input\": input_number,\n",
    "        \"is_even\": is_even_result,\n",
    "        \"operation\": operation,\n",
    "        \"result\": result\n",
    "    }\n",
    "\n",
    "# 测试不同数字\n",
    "config = {\"configurable\": {\"thread_id\": \"number_workflow_1\"}}\n",
    "\n",
    "print(\"=== 测试偶数 ===\")\n",
    "result_even = number_workflow.invoke(4, config)\n",
    "print(f\"结果: {result_even}\")\n",
    "print()\n",
    "\n",
    "print(\"=== 测试奇数 ===\")\n",
    "result_odd = number_workflow.invoke(5, config)\n",
    "print(f\"结果: {result_odd}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-8",
   "metadata": {},
   "source": [
    "**💡 控制流在 Functional API 中的应用**：\n",
    "\n",
    "- **条件分支**: 使用标准 Python `if/else` 语句实现工作流分支\n",
    "- **任务编排**: 通过 `.result()` 方法同步获取任务执行结果\n",
    "- **状态传递**: 任务结果可作为后续任务的输入\n",
    "- **灵活组合**: 可以根据业务逻辑动态组合不同的任务执行路径"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-9",
   "metadata": {},
   "source": [
    "##### 示例 6-9：使用 `invoke` 同步执行工作流\n",
    "\n",
    "演示同步工作流执行的基本模式："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "cell-9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "执行同步工作流...\n",
      "同步结果: 工作流完成: 已处理: HELLO WORLD\n",
      "调用 1: 工作流完成: 已处理: APPLE\n",
      "调用 2: 工作流完成: 已处理: BANANA\n",
      "调用 3: 工作流完成: 已处理: CHERRY\n"
     ]
    }
   ],
   "source": [
    "@task\n",
    "def process_data(data: str) -> str:\n",
    "    \"\"\"处理数据的简单任务。\"\"\"\n",
    "    time.sleep(0.1)  # 模拟处理时间\n",
    "    return f\"已处理: {data.upper()}\"\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def simple_workflow(input_data: str) -> str:\n",
    "    \"\"\"简单的数据处理工作流。\"\"\"\n",
    "    processed = process_data(input_data).result()\n",
    "    return f\"工作流完成: {processed}\"\n",
    "\n",
    "# 同步执行示例\n",
    "config = {\"configurable\": {\"thread_id\": \"sync_workflow_1\"}}\n",
    "print(\"执行同步工作流...\")\n",
    "result = simple_workflow.invoke(\"hello world\", config)\n",
    "print(f\"同步结果: {result}\")\n",
    "\n",
    "# 多次调用同一工作流\n",
    "inputs = [\"apple\", \"banana\", \"cherry\"]\n",
    "for i, data in enumerate(inputs, 1):\n",
    "    config[\"configurable\"][\"thread_id\"] = f\"sync_workflow_{i}\"\n",
    "    result = simple_workflow.invoke(data, config)\n",
    "    print(f\"调用 {i}: {result}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-9",
   "metadata": {},
   "source": [
    "**💡 同步执行的特点**：\n",
    "\n",
    "- **阻塞执行**: `invoke` 方法会等待工作流完全执行完毕\n",
    "- **简单直接**: 适合不需要实时反馈的场景\n",
    "- **错误处理**: 执行过程中的异常会直接抛出\n",
    "- **线程隔离**: 每个 `thread_id` 维护独立的状态"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-10",
   "metadata": {},
   "source": [
    "##### 示例 6-10：使用 `stream` 流式执行工作流\n",
    "\n",
    "演示流式执行，提供实时执行进度反馈："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "cell-10",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "执行流式工作流（观察执行进度）...\n",
      "==================================================\n",
      "流式更新 1: ('updates', {'step_one': '步骤1完成: 测试数据'})\n",
      "流式更新 2: ('updates', {'step_two': '步骤2完成: 步骤1完成: 测试数据'})\n",
      "流式更新 3: ('updates', {'step_three': '步骤3完成: 步骤2完成: 步骤1完成: 测试数据'})\n",
      "流式更新 4: ('updates', {'multi_step_workflow': '最终结果: 步骤3完成: 步骤2完成: 步骤1完成: 测试数据'})\n",
      "==================================================\n",
      "流式执行完成！\n"
     ]
    }
   ],
   "source": [
    "@task\n",
    "def step_one(data: str) -> str:\n",
    "    \"\"\"第一个处理步骤。\"\"\"\n",
    "    time.sleep(0.2)\n",
    "    return f\"步骤1完成: {data}\"\n",
    "\n",
    "@task\n",
    "def step_two(data: str) -> str:\n",
    "    \"\"\"第二个处理步骤。\"\"\"\n",
    "    time.sleep(0.2)\n",
    "    return f\"步骤2完成: {data}\"\n",
    "\n",
    "@task\n",
    "def step_three(data: str) -> str:\n",
    "    \"\"\"第三个处理步骤。\"\"\"\n",
    "    time.sleep(0.2)\n",
    "    return f\"步骤3完成: {data}\"\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def multi_step_workflow(input_data: str) -> str:\n",
    "    \"\"\"多步骤工作流，展示流式执行。\"\"\"\n",
    "    result1 = step_one(input_data).result()\n",
    "    result2 = step_two(result1).result()\n",
    "    result3 = step_three(result2).result()\n",
    "    \n",
    "    return f\"最终结果: {result3}\"\n",
    "\n",
    "# 流式执行示例\n",
    "config = {\"configurable\": {\"thread_id\": \"stream_workflow_1\"}}\n",
    "print(\"执行流式工作流（观察执行进度）...\")\n",
    "print(\"=\" * 50)\n",
    "\n",
    "for i, chunk in enumerate(multi_step_workflow.stream(\"测试数据\", config, stream_mode=[\"updates\"]), 1):\n",
    "    print(f\"流式更新 {i}: {chunk}\")\n",
    "    time.sleep(0.1)  # 稍微延迟以便观察\n",
    "\n",
    "print(\"=\" * 50)\n",
    "print(\"流式执行完成！\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-10",
   "metadata": {},
   "source": [
    "**💡 流式执行的优势**：\n",
    "\n",
    "- **实时反馈**: 用户可以看到工作流的执行进度\n",
    "- **改善体验**: 长时间运行的任务不会让用户感到等待\n",
    "- **早期发现**: 可以及时发现执行过程中的问题\n",
    "- **灵活监控**: 支持多种流模式（updates、messages、custom）"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-11",
   "metadata": {},
   "source": [
    "##### 示例 6-11：使用 `Command` 对象恢复工作流\n",
    "\n",
    "演示人机环路中断后的工作流恢复机制："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "cell-11",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 第一阶段：执行工作流直到中断 ===\n",
      "请求审核: 创建新用户账户\n",
      "执行更新: ('updates', {'__interrupt__': (Interrupt(value='请审核以下请求: 创建新用户账户', id='c6d55bc69a12e9beeb9c2842c724e19c'),)})\n",
      "\n",
      "=== 检查工作流状态 ===\n",
      "工作流是否已中断: True\n",
      "等待中断的任务数量: 1\n",
      "任务 1: PregelTask(id='3cf1d38b-ce07-0a90-82ad-4d6b4edc4aca', name='approval_workflow', path=('__pregel_pull', 'approval_workflow'), error=None, interrupts=(Interrupt(value='请审核以下请求: 创建新用户账户', id='c6d55bc69a12e9beeb9c2842c724e19c'),), state=None, result=None)\n",
      "\n",
      "=== 第二阶段：提供人工反馈并恢复 ===\n",
      "提供人工反馈并恢复工作流...\n",
      "请求审核: 创建新用户账户\n",
      "恢复执行: ('updates', {'require_approval': '请求已批准: 创建新用户账户'})\n",
      "恢复执行: ('updates', {'process_approved_request': '处理完成: 请求已批准: 创建新用户账户'})\n",
      "恢复执行: ('updates', {'approval_workflow': '处理完成: 请求已批准: 创建新用户账户'})\n"
     ]
    }
   ],
   "source": [
    "from langgraph.types import Command, interrupt\n",
    "\n",
    "@task\n",
    "def require_approval(request: str) -> str:\n",
    "    \"\"\"需要人工审核的任务。\"\"\"\n",
    "    print(f\"请求审核: {request}\")\n",
    "    \n",
    "    # 中断工作流，等待人工输入\n",
    "    human_feedback = interrupt(f\"请审核以下请求: {request}\")\n",
    "    \n",
    "    if human_feedback and \"批准\" in str(human_feedback):\n",
    "        return f\"请求已批准: {request}\"\n",
    "    else:\n",
    "        return f\"请求被拒绝: {request}\"\n",
    "\n",
    "@task\n",
    "def process_approved_request(approved_request: str) -> str:\n",
    "    \"\"\"处理已批准的请求。\"\"\"\n",
    "    time.sleep(0.1)\n",
    "    return f\"处理完成: {approved_request}\"\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def approval_workflow(request: str) -> str:\n",
    "    \"\"\"需要审批的工作流。\"\"\"\n",
    "    approval_result = require_approval(request).result()\n",
    "    \n",
    "    if \"已批准\" in approval_result:\n",
    "        final_result = process_approved_request(approval_result).result()\n",
    "        return final_result\n",
    "    else:\n",
    "        return f\"工作流终止: {approval_result}\"\n",
    "\n",
    "# 演示中断和恢复\n",
    "config = {\"configurable\": {\"thread_id\": \"approval_workflow_1\"}}\n",
    "request_data = \"创建新用户账户\"\n",
    "\n",
    "print(\"=== 第一阶段：执行工作流直到中断 ===\")\n",
    "try:\n",
    "    # 首次执行，将在 interrupt 处暂停\n",
    "    for chunk in approval_workflow.stream(request_data, config, stream_mode=[\"updates\"]):\n",
    "        print(f\"执行更新: {chunk}\")\n",
    "except Exception as e:\n",
    "    print(f\"工作流已中断: {e}\")\n",
    "\n",
    "# 检查工作流状态\n",
    "print(\"\\n=== 检查工作流状态 ===\")\n",
    "state = approval_workflow.get_state(config)\n",
    "print(f\"工作流是否已中断: {state.next is not None}\")\n",
    "if hasattr(state, 'tasks') and state.tasks:\n",
    "    if isinstance(state.tasks, tuple):\n",
    "        print(f\"等待中断的任务数量: {len(state.tasks)}\")\n",
    "        for i, task in enumerate(state.tasks):\n",
    "            print(f\"任务 {i+1}: {task}\")\n",
    "    elif hasattr(state.tasks, 'keys'):\n",
    "        print(f\"等待中断的任务: {list(state.tasks.keys())}\")\n",
    "    else:\n",
    "        print(f\"中断状态: {state.tasks}\")\n",
    "\n",
    "# 模拟人工审核后恢复\n",
    "print(\"\\n=== 第二阶段：提供人工反馈并恢复 ===\")\n",
    "try:\n",
    "    # 使用 Command 对象提供人工反馈\n",
    "    resume_command = Command(resume=\"批准请求\")\n",
    "    \n",
    "    print(\"提供人工反馈并恢复工作流...\")\n",
    "    for chunk in approval_workflow.stream(resume_command, config, stream_mode=[\"updates\"]):\n",
    "        print(f\"恢复执行: {chunk}\")\n",
    "    \n",
    "    # 获取最终结果\n",
    "    final_state = approval_workflow.get_state(config)\n",
    "    if hasattr(final_state, 'values') and 'return_value' in final_state.values:\n",
    "        print(f\"\\n最终结果: {final_state.values['return_value']}\")\n",
    "except Exception as e:\n",
    "    print(f\"恢复执行时出错: {e}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-11",
   "metadata": {},
   "source": [
    "**💡 Command 恢复机制解析**：\n",
    "\n",
    "- **中断函数**: `interrupt()` 可在任务中暂停工作流执行\n",
    "- **状态保存**: 中断点的状态被自动保存到 checkpointer\n",
    "- **人工介入**: 允许人类在关键决策点提供输入\n",
    "- **无缝恢复**: 使用 `Command(resume=...)` 从中断点继续执行"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-12",
   "metadata": {},
   "source": [
    "##### 示例 6-12：在 Functional API 工作流中使用 LangChain LLM 集成\n",
    "\n",
    "展示 Functional API 与 LangChain 组件的深度集成："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "cell-12",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "=== 测试查询 1 ===\n",
      "用户输入: 我今天心情很好！你能告诉我一个笑话吗？\n",
      "处理查询: 我今天心情很好！你能告诉我一个笑话吗？\n",
      "检测情感: 积极\n",
      "最终回复: 太好了！ \n",
      "\n",
      "当然可以！这里有一个轻松的小笑话：\n",
      "\n",
      "**为什么企鹅不会迷路？**  \n",
      "因为它们有“企”（启）和“鹅”（鹅），组合起来就是“启鹅”——所以它们总能找到方向！ 😄\n",
      "\n",
      "（谐音梗+动物梗，希望能让您会心一笑～）\n",
      "\n",
      "=== 测试查询 2 ===\n",
      "用户输入: 我对产品的质量很不满意\n",
      "处理查询: 我对产品的质量很不满意\n",
      "检测情感: 消极\n",
      "最终回复: 我理解您的担忧。 \n",
      "\n",
      "很抱歉听到您对产品质量不满意，这确实会让人感到困扰。为了更好地帮助您解决问题，请您提供更多详细信息，例如：\n",
      "\n",
      "1. **产品名称或型号**（如：手机、家电、电子产品等）  \n",
      "2. **购买渠道**（如：电商平台、线下门店、官网等）  \n",
      "3. **具体问题描述**（如：功能故障、外观瑕疵、包装损坏等）  \n",
      "4. **是否已联系客服**（如：是否有售后沟通记录）  \n",
      "\n",
      "根据您提供的信息，我会尽力为您提供针对性的解决方案，例如：  \n",
      "- 检查产品是否符合质量标准  \n",
      "- 指导您如何申请退换货或维修  \n",
      "- 协助联系相关客服或技术支持团队  \n",
      "\n",
      "同时，请您注意保留以下凭证：  \n",
      "- 产品照片/视频（展示问题细节）  \n",
      "- 发票或订单号  \n",
      "- 与客服沟通的记录（如聊天截图、邮件等）  \n",
      "\n",
      "我们非常重视您的反馈，期待尽快为您解决！ 🌟\n",
      "\n",
      "=== 测试查询 3 ===\n",
      "用户输入: 请问你们的营业时间是什么？\n",
      "处理查询: 请问你们的营业时间是什么？\n",
      "检测情感: 中性\n",
      "\n",
      "这句话是一个中性的问题，询问营业时间本身不带有积极或消极的情感色彩。它属于客观信息的查询，没有表达满意、不满、喜悦、愤怒等情绪。\n",
      "最终回复: 太好了！ \n",
      "\n",
      "您好！关于营业时间，我需要您提供更具体的信息，例如您想了解的是哪个店铺、商场或服务的营业时间？不同的场所可能有不同的营业安排，比如有的商场是早9点到晚9点，有的店铺可能有午休或周末休息时间。如果您能说明具体地点或服务类型，我可以更好地为您解答哦！ 😊\n",
      "\n",
      "如果无法提供具体信息，建议您直接联系相关商家或查看其官方公告，以获取最准确的营业时间。\n"
     ]
    }
   ],
   "source": [
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.func import task, entrypoint\n",
    "from langgraph.checkpoint.memory import MemorySaver\n",
    "\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen3-8B\", temperature=0.7)\n",
    "\n",
    "@task\n",
    "def generate_response(user_query: str) -> str:\n",
    "    \"\"\"生成使用 LangChain LLM 的响应的任务。\"\"\"\n",
    "    print(f\"处理查询: {user_query}\")\n",
    "    response = llm.invoke(user_query)\n",
    "    return response.content\n",
    "\n",
    "@task\n",
    "def analyze_sentiment(text: str) -> str:\n",
    "    \"\"\"分析文本情感。\"\"\"\n",
    "    sentiment_prompt = f\"分析以下文本的情感倾向，回答'积极'、'消极'或'中性'：{text}\"\n",
    "    response = llm.invoke(sentiment_prompt)\n",
    "    return response.content.strip()\n",
    "\n",
    "@task\n",
    "def enhance_response(original_response: str, sentiment: str) -> str:\n",
    "    \"\"\"根据情感分析增强响应。\"\"\"\n",
    "    if \"积极\" in sentiment:\n",
    "        enhancement = \"太好了！\"\n",
    "    elif \"消极\" in sentiment:\n",
    "        enhancement = \"我理解您的担忧。\"\n",
    "    else:\n",
    "        enhancement = \"谢谢您的问题。\"\n",
    "    \n",
    "    return f\"{enhancement} {original_response}\"\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def smart_chatbot_workflow(query: str) -> dict:\n",
    "    \"\"\"智能聊天机器人工作流，具备情感分析能力。\"\"\"\n",
    "    # 生成初始响应\n",
    "    initial_response = generate_response(query).result()\n",
    "    \n",
    "    # 分析用户查询的情感\n",
    "    user_sentiment = analyze_sentiment(query).result()\n",
    "    \n",
    "    # 根据情感增强响应\n",
    "    enhanced_response = enhance_response(initial_response, user_sentiment).result()\n",
    "    \n",
    "    return {\n",
    "        \"user_query\": query,\n",
    "        \"initial_response\": initial_response,\n",
    "        \"detected_sentiment\": user_sentiment,\n",
    "        \"final_response\": enhanced_response\n",
    "    }\n",
    "\n",
    "# 测试智能聊天机器人\n",
    "config = {\"configurable\": {\"thread_id\": \"chatbot_1\"}}\n",
    "\n",
    "# 测试不同情感的查询\n",
    "test_queries = [\n",
    "    \"我今天心情很好！你能告诉我一个笑话吗？\",\n",
    "    \"我对产品的质量很不满意\",\n",
    "    \"请问你们的营业时间是什么？\"\n",
    "]\n",
    "\n",
    "for i, query in enumerate(test_queries, 1):\n",
    "    print(f\"\\n=== 测试查询 {i} ===\")\n",
    "    print(f\"用户输入: {query}\")\n",
    "    \n",
    "    config[\"configurable\"][\"thread_id\"] = f\"chatbot_{i}\"\n",
    "    result = smart_chatbot_workflow.invoke(query, config)\n",
    "    \n",
    "    print(f\"检测情感: {result['detected_sentiment']}\")\n",
    "    print(f\"最终回复: {result['final_response']}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-12",
   "metadata": {},
   "source": [
    "**💡 LangChain 集成的价值**：\n",
    "\n",
    "- **组件复用**: 充分利用 LangChain 生态系统的现有组件\n",
    "- **模型抽象**: 通过统一接口使用不同的语言模型\n",
    "- **功能增强**: LangGraph 的状态管理增强了 LangChain 的能力\n",
    "- **生态协同**: 两个框架的优势相互补充"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-13",
   "metadata": {},
   "source": [
    "##### 示例 6-13：在 Functional API 工作流中实现任务的并行执行\n",
    "\n",
    "展示如何并行执行多个独立任务以提高性能："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cell-13",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 并行执行工作流 ===\n",
      "开始并行获取用户 12345 的数据...\n",
      "所有数据获取任务已启动，等待结果...\n",
      "所有数据已获取，开始聚合...\n",
      "\n",
      "执行耗时: 0.41 秒\n",
      "用户档案: {'user_id': '12345', 'name': 'User-12345', 'email': '12345@example.com'}\n",
      "订单总数: 2\n",
      "总消费: $249.98\n",
      "用户偏好: {'theme': 'dark', 'language': 'zh-CN', 'notifications': True}\n"
     ]
    }
   ],
   "source": [
    "@task\n",
    "def fetch_user_info(user_id: str) -> dict:\n",
    "    \"\"\"获取用户信息的任务。\"\"\"\n",
    "    time.sleep(0.3)  # 模拟 API 调用延迟\n",
    "    return {\n",
    "        \"user_id\": user_id,\n",
    "        \"name\": f\"User-{user_id}\",\n",
    "        \"email\": f\"{user_id}@example.com\"\n",
    "    }\n",
    "\n",
    "@task\n",
    "def fetch_user_orders(user_id: str) -> list:\n",
    "    \"\"\"获取用户订单的任务。\"\"\"\n",
    "    time.sleep(0.4)  # 模拟数据库查询延迟\n",
    "    return [\n",
    "        {\"order_id\": f\"ord-{user_id}-001\", \"amount\": 99.99},\n",
    "        {\"order_id\": f\"ord-{user_id}-002\", \"amount\": 149.99}\n",
    "    ]\n",
    "\n",
    "@task\n",
    "def fetch_user_preferences(user_id: str) -> dict:\n",
    "    \"\"\"获取用户偏好的任务。\"\"\"\n",
    "    time.sleep(0.2)  # 模拟缓存查询延迟\n",
    "    return {\n",
    "        \"theme\": \"dark\",\n",
    "        \"language\": \"zh-CN\",\n",
    "        \"notifications\": True\n",
    "    }\n",
    "\n",
    "@task\n",
    "def aggregate_user_data(user_info: dict, orders: list, preferences: dict) -> dict:\n",
    "    \"\"\"聚合用户数据的任务。\"\"\"\n",
    "    return {\n",
    "        \"user_profile\": user_info,\n",
    "        \"order_history\": orders,\n",
    "        \"user_preferences\": preferences,\n",
    "        \"total_orders\": len(orders),\n",
    "        \"total_spent\": sum(order[\"amount\"] for order in orders)\n",
    "    }\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def parallel_user_data_workflow(user_id: str) -> dict:\n",
    "    \"\"\"并行获取用户数据的工作流。\"\"\"\n",
    "    print(f\"开始并行获取用户 {user_id} 的数据...\")\n",
    "    \n",
    "    # 并行启动所有数据获取任务\n",
    "    user_info_future = fetch_user_info(user_id)\n",
    "    orders_future = fetch_user_orders(user_id)\n",
    "    preferences_future = fetch_user_preferences(user_id)\n",
    "    \n",
    "    print(\"所有数据获取任务已启动，等待结果...\")\n",
    "    \n",
    "    # 等待所有任务完成并获取结果\n",
    "    user_info = user_info_future.result()\n",
    "    orders = orders_future.result()\n",
    "    preferences = preferences_future.result()\n",
    "    \n",
    "    print(\"所有数据已获取，开始聚合...\")\n",
    "    \n",
    "    # 聚合数据\n",
    "    aggregated_data = aggregate_user_data(user_info, orders, preferences).result()\n",
    "    \n",
    "    return aggregated_data\n",
    "\n",
    "# 测试并行执行\n",
    "import time\n",
    "\n",
    "config = {\"configurable\": {\"thread_id\": \"parallel_workflow_1\"}}\n",
    "user_id = \"12345\"\n",
    "\n",
    "start_time = time.time()\n",
    "print(\"=== 并行执行工作流 ===\")\n",
    "result = parallel_user_data_workflow.invoke(user_id, config)\n",
    "end_time = time.time()\n",
    "\n",
    "print(f\"\\n执行耗时: {end_time - start_time:.2f} 秒\")\n",
    "print(f\"用户档案: {result['user_profile']}\")\n",
    "print(f\"订单总数: {result['total_orders']}\")\n",
    "print(f\"总消费: ${result['total_spent']:.2f}\")\n",
    "print(f\"用户偏好: {result['user_preferences']}\")\n",
    "\n",
    "# 对比：如果是串行执行会花费更长时间\n",
    "# (0.3 + 0.4 + 0.2 = 0.9秒 vs 并行执行约0.4秒)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-13",
   "metadata": {},
   "source": [
    "**💡 并行执行的性能优势**：\n",
    "\n",
    "- **时间效率**: 多个独立任务可同时执行，显著减少总耗时\n",
    "- **资源利用**: 充分利用系统的并发处理能力\n",
    "- **I/O 优化**: 特别适用于网络请求、数据库查询等 I/O 密集型任务\n",
    "- **简单实现**: 无需复杂的并发编程，框架自动处理并行执行"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-14",
   "metadata": {},
   "source": [
    "##### 示例 6-14：在 Functional API 中调用子图\n",
    "\n",
    "展示如何组合和嵌套工作流，实现模块化设计："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "cell-14",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 测试有效数据 ===\n",
      "=== 开始主数据工作流 ===\n",
      "步骤 1: 数据验证\n",
      "验证数据: ['name', 'email', 'age']\n",
      "步骤 2: 数据处理\n",
      "处理数据: {'name': 'john doe', 'email': 'John.Doe@Example.COM', 'age': 30}\n",
      "处理结果: success\n",
      "处理后数据: {'name': 'John Doe', 'email': 'john.doe@example.com', 'age': 30, 'processed_at': '2024-01-01 12:00:00'}\n",
      "\n",
      "=== 测试无效数据 ===\n",
      "=== 开始主数据工作流 ===\n",
      "步骤 1: 数据验证\n",
      "验证数据: ['email', 'age']\n",
      "处理结果: failed\n",
      "失败原因: 数据验证失败\n",
      "错误信息: ['缺少 name 字段', '邮箱格式无效']\n"
     ]
    }
   ],
   "source": [
    "# 定义子工作流\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def data_validation_workflow(data: dict) -> dict:\n",
    "    \"\"\"数据验证子工作流。\"\"\"\n",
    "    print(f\"验证数据: {list(data.keys())}\")\n",
    "    \n",
    "    validation_results = {\n",
    "        \"is_valid\": True,\n",
    "        \"errors\": [],\n",
    "        \"warnings\": []\n",
    "    }\n",
    "    \n",
    "    # 模拟数据验证逻辑\n",
    "    if \"name\" not in data:\n",
    "        validation_results[\"errors\"].append(\"缺少 name 字段\")\n",
    "        validation_results[\"is_valid\"] = False\n",
    "    \n",
    "    if \"email\" in data and \"@\" not in data[\"email\"]:\n",
    "        validation_results[\"errors\"].append(\"邮箱格式无效\")\n",
    "        validation_results[\"is_valid\"] = False\n",
    "    \n",
    "    if \"age\" in data and data[\"age\"] < 0:\n",
    "        validation_results[\"warnings\"].append(\"年龄为负数\")\n",
    "    \n",
    "    return {\n",
    "        \"original_data\": data,\n",
    "        \"validation\": validation_results\n",
    "    }\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def data_processing_workflow(data: dict) -> dict:\n",
    "    \"\"\"数据处理子工作流。\"\"\"\n",
    "    print(f\"处理数据: {data}\")\n",
    "    \n",
    "    processed_data = data.copy()\n",
    "    \n",
    "    # 模拟数据处理逻辑\n",
    "    if \"name\" in processed_data:\n",
    "        processed_data[\"name\"] = processed_data[\"name\"].title()\n",
    "    \n",
    "    if \"email\" in processed_data:\n",
    "        processed_data[\"email\"] = processed_data[\"email\"].lower()\n",
    "    \n",
    "    processed_data[\"processed_at\"] = \"2024-01-01 12:00:00\"\n",
    "    \n",
    "    return {\n",
    "        \"original_data\": data,\n",
    "        \"processed_data\": processed_data\n",
    "    }\n",
    "\n",
    "# 主工作流调用子工作流\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def master_data_workflow(raw_data: dict) -> dict:\n",
    "    \"\"\"主数据工作流，协调多个子工作流。\"\"\"\n",
    "    print(\"=== 开始主数据工作流 ===\")\n",
    "    \n",
    "    # 第一步：数据验证\n",
    "    print(\"步骤 1: 数据验证\")\n",
    "    validation_result = data_validation_workflow.invoke(raw_data)\n",
    "    \n",
    "    if not validation_result[\"validation\"][\"is_valid\"]:\n",
    "        return {\n",
    "            \"status\": \"failed\",\n",
    "            \"reason\": \"数据验证失败\",\n",
    "            \"errors\": validation_result[\"validation\"][\"errors\"],\n",
    "            \"raw_data\": raw_data\n",
    "        }\n",
    "    \n",
    "    # 第二步：数据处理（仅在验证通过时执行）\n",
    "    print(\"步骤 2: 数据处理\")\n",
    "    processing_result = data_processing_workflow.invoke(raw_data)\n",
    "    \n",
    "    return {\n",
    "        \"status\": \"success\",\n",
    "        \"validation_result\": validation_result[\"validation\"],\n",
    "        \"processed_data\": processing_result[\"processed_data\"],\n",
    "        \"original_data\": raw_data\n",
    "    }\n",
    "\n",
    "# 测试子工作流组合\n",
    "config = {\"configurable\": {\"thread_id\": \"master_workflow_1\"}}\n",
    "\n",
    "# 测试有效数据\n",
    "print(\"=== 测试有效数据 ===\")\n",
    "valid_data = {\n",
    "    \"name\": \"john doe\",\n",
    "    \"email\": \"John.Doe@Example.COM\",\n",
    "    \"age\": 30\n",
    "}\n",
    "\n",
    "result_valid = master_data_workflow.invoke(valid_data, config)\n",
    "print(f\"处理结果: {result_valid['status']}\")\n",
    "if result_valid['status'] == 'success':\n",
    "    print(f\"处理后数据: {result_valid['processed_data']}\")\n",
    "print()\n",
    "\n",
    "# 测试无效数据\n",
    "print(\"=== 测试无效数据 ===\")\n",
    "invalid_data = {\n",
    "    \"email\": \"invalid-email\",  # 无效邮箱\n",
    "    \"age\": -5  # 负数年龄\n",
    "}\n",
    "\n",
    "config[\"configurable\"][\"thread_id\"] = \"master_workflow_2\"\n",
    "result_invalid = master_data_workflow.invoke(invalid_data, config)\n",
    "print(f\"处理结果: {result_invalid['status']}\")\n",
    "if result_invalid['status'] == 'failed':\n",
    "    print(f\"失败原因: {result_invalid['reason']}\")\n",
    "    print(f\"错误信息: {result_invalid['errors']}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-14",
   "metadata": {},
   "source": [
    "**💡 子工作流组合的优势**：\n",
    "\n",
    "- **模块化设计**: 将复杂工作流分解为可重用的子工作流\n",
    "- **职责分离**: 每个子工作流专注于特定功能\n",
    "- **易于测试**: 可以独立测试每个子工作流\n",
    "- **代码复用**: 子工作流可在多个主工作流中重复使用\n",
    "- **状态隔离**: 每个工作流可以有独立的状态管理"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-15",
   "metadata": {},
   "source": [
    "##### 示例 6-15：在 Functional API 中演示自定义数据流式传输\n",
    "\n",
    "展示如何使用 `writer` 参数实现自定义数据的实时流式传输："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "cell-15",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 流式执行文本分析工作流 ===\n",
      "监听自定义流和更新流...\n",
      "--------------------------------------------------\n",
      "📢 进度更新: 开始文本分析工作流\n",
      "📢 进度更新: 输入文本长度: 98 字符\n",
      "📢 进度更新: 阶段 1: 正在分析文本结构...\n",
      "🔄 工作流更新: ['analyze_text']\n",
      "📢 进度更新: 分析完成: 发现 5 个单词\n",
      "📢 进度更新: 阶段 2: 正在生成摘要...\n",
      "🔄 工作流更新: ['generate_summary']\n",
      "📢 进度更新: 摘要生成完成\n",
      "📢 进度更新: 文本分析工作流已完成\n",
      "🔄 工作流更新: ['text_analysis_workflow']\n",
      "--------------------------------------------------\n",
      "流式执行完成！\n",
      "\n",
      "=== 最终结果 ===\n",
      "工作流执行完成，查看流式输出了解详细过程。\n"
     ]
    }
   ],
   "source": [
    "from langgraph.types import StreamWriter\n",
    "\n",
    "@task\n",
    "def analyze_text(text: str) -> dict:\n",
    "    \"\"\"分析文本的任务。\"\"\"\n",
    "    time.sleep(0.5)  # 模拟分析时间\n",
    "    \n",
    "    return {\n",
    "        \"length\": len(text),\n",
    "        \"words\": len(text.split()),\n",
    "        \"sentences\": text.count('.') + text.count('!') + text.count('?'),\n",
    "        \"paragraphs\": text.count('\\n\\n') + 1\n",
    "    }\n",
    "\n",
    "@task\n",
    "def generate_summary(analysis: dict) -> str:\n",
    "    \"\"\"生成摘要的任务。\"\"\"\n",
    "    time.sleep(0.3)  # 模拟生成时间\n",
    "    \n",
    "    return f\"文本包含 {analysis['words']} 个单词，{analysis['sentences']} 个句子，{analysis['paragraphs']} 个段落。\"\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def text_analysis_workflow(\n",
    "    text: str, \n",
    "    *,\n",
    "    writer: StreamWriter\n",
    ") -> dict:\n",
    "    \"\"\"文本分析工作流，带有实时进度反馈。\"\"\"\n",
    "    writer(\"开始文本分析工作流\")\n",
    "    writer(f\"输入文本长度: {len(text)} 字符\")\n",
    "    \n",
    "    # 第一阶段：文本分析\n",
    "    writer(\"阶段 1: 正在分析文本结构...\")\n",
    "    analysis = analyze_text(text).result()\n",
    "    writer(f\"分析完成: 发现 {analysis['words']} 个单词\")\n",
    "    \n",
    "    # 第二阶段：生成摘要\n",
    "    writer(\"阶段 2: 正在生成摘要...\")\n",
    "    summary = generate_summary(analysis).result()\n",
    "    writer(\"摘要生成完成\")\n",
    "    \n",
    "    # 最终结果\n",
    "    result = {\n",
    "        \"text_analysis\": analysis,\n",
    "        \"summary\": summary,\n",
    "        \"processing_status\": \"completed\"\n",
    "    }\n",
    "    \n",
    "    writer(\"文本分析工作流已完成\")\n",
    "    return result\n",
    "\n",
    "# 测试自定义数据流式传输\n",
    "config = {\"configurable\": {\"thread_id\": \"streaming_workflow_1\"}}\n",
    "sample_text = \"\"\"\n",
    "人工智能是一个快速发展的领域。它包含机器学习、深度学习、自然语言处理等多个分支。\n",
    "AI 技术正在改变我们的生活方式。从智能助手到自动驾驶，AI 无处不在。\n",
    "我们需要负责任地开发和使用这些技术。\n",
    "\"\"\"\n",
    "\n",
    "print(\"=== 流式执行文本分析工作流 ===\")\n",
    "print(\"监听自定义流和更新流...\")\n",
    "print(\"-\" * 50)\n",
    "\n",
    "for chunk in text_analysis_workflow.stream(\n",
    "    sample_text, \n",
    "    config, \n",
    "    stream_mode=[\"custom\", \"updates\"]\n",
    "):\n",
    "    if chunk[0] == \"custom\":\n",
    "        print(f\"📢 进度更新: {chunk[1]}\")\n",
    "    elif chunk[0] == \"updates\":\n",
    "        print(f\"🔄 工作流更新: {list(chunk[1].keys())}\")\n",
    "\n",
    "print(\"-\" * 50)\n",
    "print(\"流式执行完成！\")\n",
    "\n",
    "# 获取最终结果\n",
    "final_state = text_analysis_workflow.get_state(config)\n",
    "if hasattr(final_state, 'values') and final_state.values:\n",
    "    print(\"\\n=== 最终结果 ===\")\n",
    "    # 注意：实际的结果访问方式可能因版本而异\n",
    "    print(\"工作流执行完成，查看流式输出了解详细过程。\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-15",
   "metadata": {},
   "source": [
    "**💡 自定义流式传输的应用价值**：\n",
    "\n",
    "- **进度追踪**: 为用户提供详细的执行进度信息\n",
    "- **实时反馈**: 长时间运行的任务不再是黑盒操作\n",
    "- **调试支持**: 通过流式输出更容易定位问题\n",
    "- **用户体验**: 显著改善用户对系统响应性的感知"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-16",
   "metadata": {},
   "source": [
    "##### 示例 6-16：在 Functional API 中实现重试策略\n",
    "\n",
    "演示更复杂的重试策略配置和错误处理："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "cell-16",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "============================================================\n",
      "测试 1: https://api.example.com/data\n",
      "============================================================\n",
      "=== 开始处理 API 请求: https://api.example.com/data ===\n",
      "步骤 1: 发起网络请求\n",
      "尝试访问: https://api.example.com/data\n",
      "❌ 网络连接失败\n",
      "尝试访问: https://api.example.com/data\n",
      "❌ 网络连接失败\n",
      "尝试访问: https://api.example.com/data\n",
      "❌ 网络连接失败\n",
      "\n",
      "💥 网络错误（重试后仍失败）: 无法连接到 https://api.example.com/data\n",
      "\n",
      "📊 最终结果:\n",
      "状态: network_error\n",
      "错误信息: 无法连接到 https://api.example.com/data\n",
      "\n",
      "============================================================\n",
      "测试 2: https://api.service.com/info\n",
      "============================================================\n",
      "=== 开始处理 API 请求: https://api.service.com/info ===\n",
      "步骤 1: 发起网络请求\n",
      "尝试访问: https://api.service.com/info\n",
      "❌ 网络连接失败\n",
      "尝试访问: https://api.service.com/info\n",
      "❌ 网络连接失败\n",
      "尝试访问: https://api.service.com/info\n",
      "❌ 网络连接失败\n",
      "\n",
      "💥 网络错误（重试后仍失败）: 无法连接到 https://api.service.com/info\n",
      "\n",
      "📊 最终结果:\n",
      "状态: network_error\n",
      "错误信息: 无法连接到 https://api.service.com/info\n",
      "\n",
      "============================================================\n",
      "测试 3: https://api.test.com/status\n",
      "============================================================\n",
      "=== 开始处理 API 请求: https://api.test.com/status ===\n",
      "步骤 1: 发起网络请求\n",
      "尝试访问: https://api.test.com/status\n",
      "❌ 网络连接失败\n",
      "尝试访问: https://api.test.com/status\n",
      "❌ 网络连接失败\n",
      "尝试访问: https://api.test.com/status\n",
      "✅ 网络请求成功\n",
      "\n",
      "步骤 2: 验证响应数据\n",
      "验证数据: ['url', 'status', 'data']\n",
      "✅ 数据验证通过\n",
      "\n",
      "步骤 3: 处理验证数据\n",
      "\n",
      "📊 最终结果:\n",
      "状态: success\n",
      "处理结果: 处理完成: 来自 https://api.test.com/status 的数据\n"
     ]
    }
   ],
   "source": [
    "from langgraph.types import RetryPolicy\n",
    "import random\n",
    "\n",
    "# 定义不同的重试策略\n",
    "network_retry_policy = RetryPolicy(max_attempts=3, retry_on=ConnectionError)\n",
    "validation_retry_policy = RetryPolicy(max_attempts=2, retry_on=ValueError)\n",
    "\n",
    "@task(name=\"network_request\", retry_policy=network_retry_policy)\n",
    "def make_network_request(url: str) -> dict:\n",
    "    \"\"\"模拟网络请求，可能失败的任务。\"\"\"\n",
    "    print(f\"尝试访问: {url}\")\n",
    "    \n",
    "    # 模拟网络不稳定\n",
    "    if random.random() < 0.7:  # 70% 失败率\n",
    "        print(\"❌ 网络连接失败\")\n",
    "        raise ConnectionError(f\"无法连接到 {url}\")\n",
    "    \n",
    "    print(\"✅ 网络请求成功\")\n",
    "    return {\n",
    "        \"url\": url,\n",
    "        \"status\": \"success\",\n",
    "        \"data\": f\"来自 {url} 的数据\"\n",
    "    }\n",
    "\n",
    "@task(name=\"data_validator\", retry_policy=validation_retry_policy)\n",
    "def validate_response_data(data: dict) -> dict:\n",
    "    \"\"\"验证响应数据，可能需要重试的任务。\"\"\"\n",
    "    print(f\"验证数据: {list(data.keys())}\")\n",
    "    \n",
    "    # 模拟验证逻辑\n",
    "    if \"status\" not in data:\n",
    "        print(\"❌ 数据验证失败：缺少 status 字段\")\n",
    "        raise ValueError(\"响应数据格式不正确\")\n",
    "    \n",
    "    if data.get(\"status\") != \"success\":\n",
    "        print(\"❌ 数据验证失败：状态不是 success\")\n",
    "        raise ValueError(\"响应状态异常\")\n",
    "    \n",
    "    print(\"✅ 数据验证通过\")\n",
    "    return {\n",
    "        \"validated\": True,\n",
    "        \"original_data\": data\n",
    "    }\n",
    "\n",
    "@task\n",
    "def process_validated_data(validated_data: dict) -> str:\n",
    "    \"\"\"处理已验证的数据。\"\"\"\n",
    "    original = validated_data[\"original_data\"]\n",
    "    return f\"处理完成: {original['data']}\"\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def robust_api_workflow(api_url: str) -> dict:\n",
    "    \"\"\"具有重试机制的健壮 API 工作流。\"\"\"\n",
    "    try:\n",
    "        print(f\"=== 开始处理 API 请求: {api_url} ===\")\n",
    "        \n",
    "        # 步骤 1: 网络请求（带重试）\n",
    "        print(\"步骤 1: 发起网络请求\")\n",
    "        response_data = make_network_request(api_url).result()\n",
    "        \n",
    "        # 步骤 2: 数据验证（带重试）\n",
    "        print(\"\\n步骤 2: 验证响应数据\")\n",
    "        validated_data = validate_response_data(response_data).result()\n",
    "        \n",
    "        # 步骤 3: 处理数据（无重试）\n",
    "        print(\"\\n步骤 3: 处理验证数据\")\n",
    "        processed_result = process_validated_data(validated_data).result()\n",
    "        \n",
    "        return {\n",
    "            \"status\": \"success\",\n",
    "            \"result\": processed_result,\n",
    "            \"api_url\": api_url\n",
    "        }\n",
    "        \n",
    "    except ConnectionError as e:\n",
    "        print(f\"\\n💥 网络错误（重试后仍失败）: {e}\")\n",
    "        return {\n",
    "            \"status\": \"network_error\",\n",
    "            \"error\": str(e),\n",
    "            \"api_url\": api_url\n",
    "        }\n",
    "        \n",
    "    except ValueError as e:\n",
    "        print(f\"\\n💥 数据验证错误（重试后仍失败）: {e}\")\n",
    "        return {\n",
    "            \"status\": \"validation_error\",\n",
    "            \"error\": str(e),\n",
    "            \"api_url\": api_url\n",
    "        }\n",
    "        \n",
    "    except Exception as e:\n",
    "        print(f\"\\n💥 未预期错误: {e}\")\n",
    "        return {\n",
    "            \"status\": \"unexpected_error\",\n",
    "            \"error\": str(e),\n",
    "            \"api_url\": api_url\n",
    "        }\n",
    "\n",
    "# 测试重试机制\n",
    "api_endpoints = [\n",
    "    \"https://api.example.com/data\",\n",
    "    \"https://api.service.com/info\",\n",
    "    \"https://api.test.com/status\"\n",
    "]\n",
    "\n",
    "for i, endpoint in enumerate(api_endpoints, 1):\n",
    "    print(f\"\\n{'='*60}\")\n",
    "    print(f\"测试 {i}: {endpoint}\")\n",
    "    print(f\"{'='*60}\")\n",
    "    \n",
    "    config = {\"configurable\": {\"thread_id\": f\"robust_api_workflow_{i}\"}}\n",
    "    result = robust_api_workflow.invoke(endpoint, config)\n",
    "    \n",
    "    print(\"\\n📊 最终结果:\")\n",
    "    print(f\"状态: {result['status']}\")\n",
    "    if result['status'] == 'success':\n",
    "        print(f\"处理结果: {result['result']}\")\n",
    "    else:\n",
    "        print(f\"错误信息: {result['error']}\")\n",
    "    \n",
    "    time.sleep(0.5)  # 给用户时间观察输出"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-16",
   "metadata": {},
   "source": [
    "**💡 重试策略的重要性**：\n",
    "\n",
    "- **提高可靠性**: 自动处理临时性错误，减少系统故障\n",
    "- **区分异常类型**: 不同类型的错误采用不同的重试策略\n",
    "- **优雅降级**: 重试失败后提供有意义的错误信息\n",
    "- **生产就绪**: 是构建健壮生产系统的必要功能"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-6-17",
   "metadata": {},
   "source": [
    "##### 示例 6-17：在 Functional API 中实现短期记忆\n",
    "\n",
    "展示如何使用 `previous` 参数和 `entrypoint.final()` 管理工作流状态："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "cell-17",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 测试计数器工作流 ===\n",
      "当前计数: 0, 增量: 1, 新计数: 1\n",
      "调用 1: 增量 1 -> 返回值 0\n",
      "当前计数: 1, 增量: 2, 新计数: 3\n",
      "调用 2: 增量 2 -> 返回值 1\n",
      "当前计数: 3, 增量: 3, 新计数: 6\n",
      "调用 3: 增量 3 -> 返回值 3\n",
      "当前计数: 6, 增量: 5, 新计数: 11\n",
      "调用 4: 增量 5 -> 返回值 6\n",
      "\n",
      "=== 测试会话聊天工作流 ===\n",
      "\n",
      "--- 对话轮次 1 ---\n",
      "👤 用户: 你好！\n",
      "🆕 开始新的对话会话\n",
      "🤖 助手: 这是我们的第 1 次交流。你好！\n",
      "\n",
      "--- 对话轮次 2 ---\n",
      "👤 用户: 我叫小明\n",
      "🔄 恢复对话会话 (已有 2 条消息)\n",
      "🤖 助手: 很高兴认识你，小明！\n",
      "\n",
      "--- 对话轮次 3 ---\n",
      "👤 用户: 今天天气怎么样？\n",
      "🔄 恢复对话会话 (已有 4 条消息)\n",
      "🤖 助手: 小明，这是我们的第 3 次交流。你说：今天天气怎么样？\n",
      "\n",
      "--- 对话轮次 4 ---\n",
      "👤 用户: 我喜欢编程\n",
      "🔄 恢复对话会话 (已有 6 条消息)\n",
      "🤖 助手: 小明，这是我们的第 4 次交流。你说：我喜欢编程\n",
      "\n",
      "--- 对话轮次 5 ---\n",
      "👤 用户: 再见！\n",
      "🔄 恢复对话会话 (已有 8 条消息)\n",
      "🤖 助手: 小明，这是我们的第 5 次交流。你说：再见！\n",
      "\n",
      "=== 查看最终会话状态 ===\n",
      "会话已保存，包含详细的对话历史\n"
     ]
    }
   ],
   "source": [
    "from typing import Any\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def counter_workflow(increment: int, *, previous: Any = None) -> 'entrypoint.final[int, int]':\n",
    "    \"\"\"使用先前状态维护计数器的工作流。\"\"\"\n",
    "    current_count = previous if previous is not None else 0\n",
    "    new_count = current_count + increment\n",
    "    \n",
    "    print(f\"当前计数: {current_count}, 增量: {increment}, 新计数: {new_count}\")\n",
    "    \n",
    "    # 返回当前值给调用者，保存新值到状态\n",
    "    return entrypoint.final(value=current_count, save=new_count)\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver())\n",
    "def session_chat_workflow(\n",
    "    user_message: str, \n",
    "    *, \n",
    "    previous: Any = None\n",
    ") -> 'entrypoint.final[str, dict]':\n",
    "    \"\"\"带有会话记忆的聊天工作流。\"\"\"\n",
    "    # 初始化或恢复会话状态\n",
    "    if previous is None:\n",
    "        session_state = {\n",
    "            \"messages\": [],\n",
    "            \"user_name\": None,\n",
    "            \"conversation_count\": 0\n",
    "        }\n",
    "        print(\"🆕 开始新的对话会话\")\n",
    "    else:\n",
    "        session_state = previous.copy()\n",
    "        print(f\"🔄 恢复对话会话 (已有 {len(session_state['messages'])} 条消息)\")\n",
    "    \n",
    "    # 更新会话状态\n",
    "    session_state[\"conversation_count\"] += 1\n",
    "    session_state[\"messages\"].append({\"role\": \"user\", \"content\": user_message})\n",
    "    \n",
    "    # 简单的聊天逻辑\n",
    "    if \"我叫\" in user_message or \"我是\" in user_message:\n",
    "        # 尝试提取用户名\n",
    "        parts = user_message.replace(\"我叫\", \"\").replace(\"我是\", \"\").strip()\n",
    "        if parts:\n",
    "            session_state[\"user_name\"] = parts\n",
    "            response = f\"很高兴认识你，{parts}！\"\n",
    "        else:\n",
    "            response = \"很高兴认识你！\"\n",
    "    elif session_state[\"user_name\"]:\n",
    "        response = f\"{session_state['user_name']}，这是我们的第 {session_state['conversation_count']} 次交流。你说：{user_message}\"\n",
    "    else:\n",
    "        response = f\"这是我们的第 {session_state['conversation_count']} 次交流。你好！\"\n",
    "    \n",
    "    # 添加助手回复到会话历史\n",
    "    session_state[\"messages\"].append({\"role\": \"assistant\", \"content\": response})\n",
    "    \n",
    "    return entrypoint.final(value=response, save=session_state)\n",
    "\n",
    "# 测试计数器工作流\n",
    "print(\"=== 测试计数器工作流 ===\")\n",
    "config_counter = {\"configurable\": {\"thread_id\": \"counter_workflow_1\"}}\n",
    "\n",
    "for i, increment in enumerate([1, 2, 3, 5], 1):\n",
    "    result = counter_workflow.invoke(increment, config_counter)\n",
    "    print(f\"调用 {i}: 增量 {increment} -> 返回值 {result}\")\n",
    "print()\n",
    "\n",
    "# 测试会话聊天工作流\n",
    "print(\"=== 测试会话聊天工作流 ===\")\n",
    "config_chat = {\"configurable\": {\"thread_id\": \"chat_workflow_1\"}}\n",
    "\n",
    "conversation = [\n",
    "    \"你好！\",\n",
    "    \"我叫小明\",\n",
    "    \"今天天气怎么样？\",\n",
    "    \"我喜欢编程\",\n",
    "    \"再见！\"\n",
    "]\n",
    "\n",
    "for i, message in enumerate(conversation, 1):\n",
    "    print(f\"\\n--- 对话轮次 {i} ---\")\n",
    "    print(f\"👤 用户: {message}\")\n",
    "    response = session_chat_workflow.invoke(message, config_chat)\n",
    "    print(f\"🤖 助手: {response}\")\n",
    "\n",
    "# 查看最终会话状态\n",
    "print(\"\\n=== 查看最终会话状态 ===\")\n",
    "final_state = session_chat_workflow.get_state(config_chat)\n",
    "if hasattr(final_state, 'values'):\n",
    "    print(\"会话已保存，包含详细的对话历史\")\n",
    "else:\n",
    "    print(\"状态管理正常工作\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-6-17",
   "metadata": {},
   "source": [
    "**💡 短期记忆机制解析**：\n",
    "\n",
    "- **`previous` 参数**: 自动注入上次执行保存的状态\n",
    "- **`entrypoint.final()`**: 分离返回值和保存状态，提供灵活的状态管理\n",
    "- **状态持续性**: 同一 `thread_id` 下的状态在调用间保持连续\n",
    "- **应用场景**: 适用于需要跨调用维护上下文的对话系统"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5c578f05",
   "metadata": {},
   "source": [
    "##### 示例 6-18：在 Functional API 中实现长期记忆"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "664d6d55",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== 使用 invoke 同步执行 ===\n",
      "结果: 您好！根据您的查询 '今天天气怎么样？' 和偏好设置，为您提供中文服务。\n",
      "\n",
      "=== 使用 stream 流式执行 ===\n",
      "📢 进度: 开始个性化工作流，线程: personalized_workflow_2\n",
      "📢 进度: 获取用户 user_456 的偏好...\n",
      "🔄 更新: ['retrieve_user_preferences']\n",
      "📢 进度: 用户偏好: {'theme': 'light', 'language': 'zh-CN', 'notifications': True}\n",
      "🔄 更新: ['store_user_preferences']\n",
      "📢 进度: 偏好更新结果: 已保存用户 user_456 的偏好设置\n",
      "📢 进度: 个性化工作流完成\n",
      "🔄 更新: ['personalized_workflow']\n",
      "\n",
      "=== 测试存储持久性 ===\n",
      "第二次调用结果: 您好！根据您的查询 '明天会下雨吗？' 和偏好设置，为您提供中文服务。\n"
     ]
    }
   ],
   "source": [
    "from langgraph.func import entrypoint, task\n",
    "from langgraph.checkpoint.memory import MemorySaver\n",
    "from langgraph.store.memory import InMemoryStore\n",
    "from langgraph.store.base import BaseStore\n",
    "from langgraph.types import StreamWriter\n",
    "from langchain_core.runnables import RunnableConfig\n",
    "from typing import Any\n",
    "\n",
    "# 初始化长期内存存储\n",
    "store = InMemoryStore()\n",
    "\n",
    "@task\n",
    "def retrieve_user_preferences(user_id: str, *, store: BaseStore):\n",
    "    \"\"\"从长期内存中检索用户偏好的任务\"\"\"\n",
    "    try:\n",
    "        preferences_data = store.get((\"user_preferences\", user_id), \"preferences\")\n",
    "        if preferences_data:\n",
    "            return preferences_data.value\n",
    "        else:\n",
    "            return {\"theme\": \"light\", \"language\": \"zh-CN\", \"notifications\": True}  # 默认偏好\n",
    "    except Exception as e:\n",
    "        print(f\"获取用户偏好时出错: {e}\")\n",
    "        return {\"theme\": \"light\", \"language\": \"zh-CN\", \"notifications\": True}\n",
    "\n",
    "@task  \n",
    "def store_user_preferences(user_id: str, preferences: dict, *, store: BaseStore):\n",
    "    \"\"\"存储用户偏好到长期内存的任务\"\"\"\n",
    "    try:\n",
    "        store.put((\"user_preferences\", user_id), \"preferences\", preferences)\n",
    "        return f\"已保存用户 {user_id} 的偏好设置\"\n",
    "    except Exception as e:\n",
    "        print(f\"存储用户偏好时出错: {e}\")\n",
    "        return f\"保存失败: {e}\"\n",
    "\n",
    "@entrypoint(checkpointer=MemorySaver(), store=store)\n",
    "def personalized_workflow(\n",
    "    input_data: dict,  \n",
    "    *,\n",
    "    store: BaseStore,\n",
    "    writer: StreamWriter,\n",
    "    config: RunnableConfig\n",
    ") -> str:\n",
    "    \"\"\"访问长期记忆的个性化工作流\"\"\"\n",
    "    # 从输入数据中提取参数\n",
    "    user_id = input_data[\"user_id\"]\n",
    "    query = input_data[\"query\"]\n",
    "\n",
    "    thread_id = config.get(\"configurable\", {}).get(\"thread_id\", \"unknown\")\n",
    "    writer(f\"开始个性化工作流，线程: {thread_id}\")\n",
    "\n",
    "    # 获取用户偏好\n",
    "    writer(f\"获取用户 {user_id} 的偏好...\")\n",
    "    user_prefs = retrieve_user_preferences(user_id, store=store).result()\n",
    "    writer(f\"用户偏好: {user_prefs}\")\n",
    "\n",
    "    # 根据偏好处理查询\n",
    "    if user_prefs.get(\"language\") == \"zh-CN\":\n",
    "        response = f\"您好！根据您的查询 '{query}' 和偏好设置，为您提供中文服务。\"\n",
    "    else:\n",
    "        response = f\"Hello! Based on your query '{query}' and preferences, providing service in English.\"\n",
    "\n",
    "    # 可选：更新偏好（例如记录最近查询）\n",
    "    updated_prefs = user_prefs.copy()\n",
    "    updated_prefs[\"last_query\"] = query\n",
    "    store_result = store_user_preferences(user_id, updated_prefs, store=store).result()\n",
    "    writer(f\"偏好更新结果: {store_result}\")\n",
    "\n",
    "    writer(\"个性化工作流完成\")\n",
    "    return response\n",
    "\n",
    "\n",
    "config = {\"configurable\": {\"thread_id\": \"personalized_workflow_1\"}}\n",
    "\n",
    "# 方式1: 使用 invoke 同步执行\n",
    "print(\"=== 使用 invoke 同步执行 ===\")\n",
    "input_data = {\n",
    "    \"user_id\": \"user_123\",\n",
    "    \"query\": \"今天天气怎么样？\"\n",
    "}\n",
    "result = personalized_workflow.invoke(input_data, config=config)\n",
    "print(f\"结果: {result}\")\n",
    "print()\n",
    "\n",
    "# 方式2: 使用 stream 流式执行（可以看到中间过程）\n",
    "print(\"=== 使用 stream 流式执行 ===\")\n",
    "input_data2 = {\n",
    "    \"user_id\": \"user_456\",\n",
    "    \"query\": \"What's the weather like today?\"\n",
    "}\n",
    "for chunk in personalized_workflow.stream(\n",
    "    input_data2,\n",
    "    config={\"configurable\": {\"thread_id\": \"personalized_workflow_2\"}},\n",
    "    stream_mode=[\"custom\", \"updates\"]\n",
    "):\n",
    "    if chunk[0] == \"custom\":\n",
    "        print(f\"📢 进度: {chunk[1]}\")\n",
    "    elif chunk[0] == \"updates\":\n",
    "        print(f\"🔄 更新: {list(chunk[1].keys())}\")\n",
    "print()\n",
    "\n",
    "# 方式3: 测试存储持久性 - 再次调用同一用户\n",
    "print(\"=== 测试存储持久性 ===\")\n",
    "input_data3 = {\n",
    "    \"user_id\": \"user_123\",  # 同一用户\n",
    "    \"query\": \"明天会下雨吗？\"\n",
    "}\n",
    "result2 = personalized_workflow.invoke(input_data3, config={\"configurable\": {\"thread_id\": \"personalized_workflow_3\"}})\n",
    "print(f\"第二次调用结果: {result2}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "api-comparison",
   "metadata": {},
   "source": [
    "### API 特性对比总结\n",
    "\n",
    "| 特性 | `create_react_agent` | Functional API | Graph API |\n",
    "|------|---------------------|----------------|----------|\n",
    "| **学习曲线** | 最低 | 中等 | 较高 |\n",
    "| **开发速度** | 最快 | 快速 | 中等 |\n",
    "| **灵活性** | 有限 | 高 | 最高 |\n",
    "| **可扩展性** | 低 | 中等 | 高 |\n",
    "| **适用场景** | ReAct 模式 | 中等复杂度 | 复杂系统 |\n",
    "| **代码风格** | 配置式 | 函数式 | 声明式 |\n",
    "| **调试支持** | LangSmith 自动 | 任务级跟踪 | 节点级可视化 |\n",
    "| **状态管理** | 自动 | 灵活 | 完全控制 |"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "chapter-conclusion",
   "metadata": {},
   "source": [
    "## 📚 本章总结\n",
    "\n",
    "通过本章的学习，我们全面探索了 LangGraph 框架提供的三种主要 API 选项，深入理解了它们各自的特点、优势和适用场景。首先学习了 `create_react_agent` 预构建 API，掌握了快速构建 ReAct 智能体的方法，包括自定义提示、记忆集成、人机环路和结构化输出等高级功能。接着深入研究了 Functional API，学会使用 `@entrypoint` 和 `@task` 装饰器构建函数式工作流，掌握了并行执行、子工作流组合、自定义流式传输、重试策略和状态管理等核心技术。然后回顾了 Graph API 的核心价值，理解了其在处理复杂智能体架构中的不可替代性。最后建立了 API 选择决策框架，学会根据项目复杂度、团队经验、时间要求和可扩展性需求选择最适合的 API。这些多样化的 API 选项确保了 LangGraph 能够满足从快速原型到复杂生产系统的各种开发需求，为构建下一代 AI 智能体应用提供了强有力的工具支持。"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.13.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
